package com.zhang.flink.exercise;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @title: 使用底层函数模拟滑动窗口
 * @author: zhangyifan
 * @date: 2022/10/13 09:19
 */
public class SlidingWinKeyedProcess {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 处理逻辑
        env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Tuple2<Integer, Long>>() {
                    @Override
                    public Tuple2<Integer, Long> map(String value) throws Exception {
                        String[] fields = value.split(" ");
                        return Tuple2.of(Integer.parseInt(fields[0]), Long.parseLong(fields[1]) * 1000L);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<Integer, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<Integer, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<Integer, Long> element, long recordTimestamp) {
                                        return element.f1;
                                    }
                                })
                )
                .process(new ProcessFunction<Tuple2<Integer, Long>, String>() {
                    private long slidingWinSize = 5 * 1000L;
                    private long slidingWinStep = 1 * 1000L;
                    // 定义状态  key:窗口开始时间 value:属于此窗口的所有element
                    private MapState<Long, List<Integer>> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(
                                new MapStateDescriptor<Long, List<Integer>>(
                                        "map-state",
                                        Types.LONG,
                                        Types.LIST(Types.INT)
                                )
                        );
                    }

                    @Override
                    public void processElement(Tuple2<Integer, Long> value, ProcessFunction<Tuple2<Integer, Long>, String>.Context ctx, Collector<String> out) throws Exception {


                    }

                    @Override
                    public void onTimer(long timestamp, ProcessFunction<Tuple2<Integer, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

                    }
                });

        // 执行任务
        env.execute();
    }
}
